
package rafthttp

import (
	"context"
	"net/http"
	"sync"
	"time"

	"go.etcd.io/etcd/client/pkg/v3/transport"
	"go.etcd.io/etcd/client/pkg/v3/types"
	"go.etcd.io/etcd/raft/v3"
	"go.etcd.io/etcd/raft/v3/raftpb"
	"go.etcd.io/etcd/server/v3/etcdserver/api/snap"
	stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"

	"github.com/xiang90/probing"
	"go.uber.org/zap"
	"golang.org/x/time/rate"
)

type Raft interface {
	// 将指定的消息实例传递到底层的etcd-raft模块进行处理
	Process(ctx context.Context, m raftpb.Message) error
	// 检测指定节点是否从当前集群中移出
	IsIDRemoved(id uint64) bool
	// 通知底层etcd-raft模块，当前节点与指定的节点无法联通
	ReportUnreachable(id uint64)
	// 通知底层的etcd-raft模块，快照数据是否发送成功
	ReportSnapshot(id uint64, status raft.SnapshotStatus)
}

// 定义了etcd网络层的核心功能
type Transporter interface {
	Start() error	// 初始化操作
	Handler() http.Handler	// 创建Handler实例，并关联到指定的URL上
	Send(m []raftpb.Message)	// 发送消息
	SendSnapshot(m snap.Message)	// 发送快照数据
	// 在集群中添加一个节点时，其他节点会通过该方法添加该新加入节点的信息
	AddRemote(id types.ID, urls []string)
	// Peer接口是当前节点对集群中其他节点的抽象表示，而结构体peer则是Peer接口的一个具体实现
	AddPeer(id types.ID, urls []string)
	RemovePeer(id types.ID)
	RemoveAllPeers()
	UpdatePeer(id types.ID, urls []string)
	ActiveSince(id types.ID) time.Time
	ActivePeers() int
	Stop()	// 关闭操作，该方法会关闭全部的网络连接
}

// 传输实现传输接口。它提供功能
// 向对等方发送raft消息，并从对等方接收raft消息。
// 用户应该调用Handler方法来获取一个处理程序来处理从PeerURL接收到的请求。
// 用户需要在调用其他函数之前调用Start，当传输不再使用时调用
// Stop。
type Transport struct {
	Logger *zap.Logger
	DialTimeout time.Duration // 请求拨出超时前的最长持续时间
	DialRetryFrequency rate.Limit
	TLSInfo transport.TLSInfo // 创建连接时使用的TLS信息
	ID          types.ID   // 当前节点自己的ID。
	URLs        types.URLs // 当前节点与集群中其他节点交互时使用的URL地址。
	ClusterID   types.ID   // 当前节点所在的集群的ID。
	// Raft是一个接口，其实现的底层封装了etcd-raft模块，
	// 当rafthttp.Transport收到消息之后，会将其交给Raft实例进行处理。
	Raft        Raft
	Snapshotter *snap.Snapshotter	//Snapshotter负责管理快照文件
	ServerStats *stats.ServerStats // 用于记录一般传输统计
	LeaderStats *stats.LeaderStats
	ErrorC chan error
	streamRt   http.RoundTripper // Stream消息通道中使用的http.RoundTripper实例。
	pipelineRt http.RoundTripper // Pipeline消息通道中使用的http.RoundTripper实例，
	mu      sync.RWMutex         // 保护远程和对等映射
	// remote中只封装了pipeline实例，remote主要负责发送快照数据，帮助新加入的节点快速追赶上其他节点的数据。
	remotes map[types.ID]*remote //
	// Peer接口是当前节点对集群中其他节点的抽象表示。对于当前节点来说，
	// 集群中其他节点在本地都会有一个Peer实例与之对应，peers字段维护了节点ID到对应Peer实例之间的映射关系。
	peers   map[types.ID]Peer    //
	pipelineProber probing.Prober	// 用于探测Pipeline消息通道是否可用。
	streamProber   probing.Prober	// 用于探测stream消息通道是否可用。
}

func (t *Transport) Start() error {
	var err error
	// 创建Stream消息通道使用的http.RoundTripper实例，底层实际上是创建http.Transport实例，
	// 这里需要读者注意一下几个参数的设置，分别是：创建连接的超时时间（根据配置指定）、
	// 读写请求的超时时间（默认Ss）和keepAl工ve时间（默认为30s)
	t.streamRt, err = newStreamRoundTripper(t.TLSInfo, t.DialTimeout)
	if err != nil {
		return err
	}
	// 创建Pipeline消息通道使用的http.RoundTripper实例
	// 与streamRt不同的是，读写请求的起时时间设置成了永不过期
	t.pipelineRt, err = NewRoundTripper(t.TLSInfo, t.DialTimeout)
	if err != nil {
		return err
	}
	//初始化remotes和peers两个map字段
	t.remotes = make(map[types.ID]*remote)
	t.peers = make(map[types.ID]Peer)
	// 该prober实例主要用于探测Pipeline消息通道是否可用
	t.pipelineProber = probing.NewProber(t.pipelineRt)
	t.streamProber = probing.NewProber(t.streamRt)
	if t.DialRetryFrequency == 0 {
		t.DialRetryFrequency = rate.Every(100 * time.Millisecond)
	}
	return nil
}

// Transport.Handler（）方法主要负责创建Steam消息通道和Pipeline消息通道用到的Handler实例，并注册到相应的请求路径上
func (t *Transport) Handler() http.Handler {
	// 创建pipelineHandler、streamHandler和snapshotHandler三个实例，这三个实例都实现／／了http.Server.Handler接口
	pipelineHandler := newPipelineHandler(t, t.Raft, t.ClusterID)
	streamHandler := newStreamHandler(t, t, t.Raft, t.ID, t.ClusterID)
	snapHandler := newSnapshotHandler(t, t.Raft, t.Snapshotter, t.ClusterID)
	// 创建ServeMux实例，ServeMux是一个多路复用器，
	// ServeMux主要通过其m字段(map[string)muxEntry类型）存储具体的URL和Handler实例之间的映射关系，
	mux := http.NewServeMux()
	// 下面就是设立URL与Handler实例的映射关系，也就是访问指定URL地址的请求，由对应的Handler实例进行处理
	mux.Handle(RaftPrefix, pipelineHandler)	// ”/raft"-> pipelineHandler
	mux.Handle(RaftStreamPrefix+"/", streamHandler)	// ”/raft/stream/”--> strearnHandler
	mux.Handle(RaftSnapshotPrefix, snapHandler)	// ”/raft/snapshot"--> snapHandler
	mux.Handle(ProbingPrefix, probing.NewHandler())	// "/lraft/probing”--> httpHealth
	return mux
}

// Get（）方法用于获取指定节点对应的Peer实例
func (t *Transport) Get(id types.ID) Peer {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.peers[id]
}

// Transport. Send（）方法负责发送指定的ra位pb.Message消息，其中首先尝试使用目标节点对应的Peer实例发送消息，
// 如果没有找到对应的Peer实例，则尝试使用对应的remote实例发送消息。
func (t *Transport) Send(msgs []raftpb.Message) {
	for _, m := range msgs {	// 遍历msgs切片中的全部消息
		if m.To == 0 {
			// 忽略故意丢弃的消息
			continue
		}
		// 根据raftpb.Message.To字段，获取目标节点对应的Peer实例
		to := types.ID(m.To)

		t.mu.RLock()
		p, pok := t.peers[to]
		g, rok := t.remotes[to]
		t.mu.RUnlock()

		if pok {	// 如果存在对应的Peer实例，则使用Peer发送消息
			if m.Type == raftpb.MsgApp {	// 统计信息
				t.ServerStats.SendAppendReq(m.Size())
			}
			p.send(m)	// 通过peer.send（）方法完成消息的发送
			continue
		}

		if rok {	// 如果指定节点ID不存在对应的Peer实例，则尝试使用查找对应remote实例
			g.send(m)	// 通过remote.send（）方法完成消息的发送
			continue
		}

		if t.Logger != nil {	// 无法找到raftpb.Message的目标节点，则记录日志信息
			t.Logger.Debug(
				"ignored message send request; unknown remote peer target",
				zap.String("type", m.Type.String()),
				zap.String("unknown-target-peer-id", to.String()),
			)
		}
	}
}

func (t *Transport) Stop() {
	t.mu.Lock()
	defer t.mu.Unlock()
	for _, r := range t.remotes {
		r.stop()
	}
	for _, p := range t.peers {
		p.stop()
	}
	t.pipelineProber.RemoveAll()
	t.streamProber.RemoveAll()
	if tr, ok := t.streamRt.(*http.Transport); ok {
		tr.CloseIdleConnections()
	}
	if tr, ok := t.pipelineRt.(*http.Transport); ok {
		tr.CloseIdleConnections()
	}
	t.peers = nil
	t.remotes = nil
}

// CutPeer将消息丢弃到指定的对等方。
func (t *Transport) CutPeer(id types.ID) {
	t.mu.RLock()
	p, pok := t.peers[id]
	g, gok := t.remotes[id]
	t.mu.RUnlock()

	if pok {
		p.(Pausable).Pause()
	}
	if gok {
		g.Pause()
	}
}

// MendPeer恢复给定对等方的消息丢弃行为。
func (t *Transport) MendPeer(id types.ID) {
	t.mu.RLock()
	p, pok := t.peers[id]
	g, gok := t.remotes[id]
	t.mu.RUnlock()

	if pok {
		p.(Pausable).Resume()
	}
	if gok {
		g.Resume()
	}
}

func (t *Transport) AddRemote(id types.ID, us []string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.remotes == nil {
		// 没有干净的方法关闭golang http服务器
		// （请参阅：https:
		// 停止传输；忽略任何新连接。
		return
	}
	if _, ok := t.peers[id]; ok {
		return
	}
	if _, ok := t.remotes[id]; ok {
		return
	}
	urls, err := types.NewURLs(us)
	if err != nil {
		if t.Logger != nil {
			t.Logger.Panic("failed NewURLs", zap.Strings("urls", us), zap.Error(err))
		}
	}
	t.remotes[id] = startRemote(t, urls, id)

	if t.Logger != nil {
		t.Logger.Info(
			"added new remote peer",
			zap.String("local-member-id", t.ID.String()),
			zap.String("remote-peer-id", id.String()),
			zap.Strings("remote-peer-urls", us),
		)
	}
}

// 创建井启动对应节点的Peer实例，
func (t *Transport) AddPeer(id types.ID, us []string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	// 检测当前rafthttp.Transport的运行状态
	if t.peers == nil {
		panic("transport stopped")
	}
	// 是否已经与指定ID的节点建立了连接
	if _, ok := t.peers[id]; ok {
		return
	}
	// 解析us切片中指定URL连接
	urls, err := types.NewURLs(us)
	if err != nil {
		if t.Logger != nil {
			t.Logger.Panic("failed NewURLs", zap.Strings("urls", us), zap.Error(err))
		}
	}
	fs := t.LeaderStats.Follower(id.String())
	// 创建指定节点对应的Peer实例，其中会相关的Stream消息通道和Pipeline消息通道
	t.peers[id] = startPeer(t, urls, id, fs)
	// 每隔一段时间，prober会向该节点发送探测消息，检测对端的健康状况
	addPeerToProber(t.Logger, t.pipelineProber, id.String(), us, RoundTripperNameSnapshot, rttSec)
	addPeerToProber(t.Logger, t.streamProber, id.String(), us, RoundTripperNameRaftMessage, rttSec)

	if t.Logger != nil {
		t.Logger.Info(
			"added remote peer",
			zap.String("local-member-id", t.ID.String()),
			zap.String("remote-peer-id", id.String()),
			zap.Strings("remote-peer-urls", us),
		)
	}
}

// RemovePeer（）方法会调用peer.stop（）方法关闭底层的连接，同时会停止定时发送的探测消息，
func (t *Transport) RemovePeer(id types.ID) {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.removePeer(id)
}

// RemovePeer（）方法会调用peer.stop（）方法关闭底层的连接，同时会停止定时发送的探测消息，
func (t *Transport) RemoveAllPeers() {
	t.mu.Lock()
	defer t.mu.Unlock()
	for id := range t.peers {
		t.removePeer(id)
	}
}

// 此函数的调用方必须具有对等互斥体。
func (t *Transport) removePeer(id types.ID) {
	if peer, ok := t.peers[id]; ok {
		peer.stop()
	} else {
		if t.Logger != nil {
			t.Logger.Panic("unexpected removal of unknown remote peer", zap.String("remote-peer-id", id.String()))
		}
	}
	delete(t.peers, id)
	delete(t.LeaderStats.Followers, id.String())
	t.pipelineProber.Remove(id.String())
	t.streamProber.Remove(id.String())

	if t.Logger != nil {
		t.Logger.Info(
			"removed remote peer",
			zap.String("local-member-id", t.ID.String()),
			zap.String("removed-remote-peer-id", id.String()),
		)
	}
}

// UpdatePeer（）方法则会调用严er.update（）方法更新对端暴露的URL地址，同时更新探测消息发送的目标地址
func (t *Transport) UpdatePeer(id types.ID, us []string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	// TODO:返回错误或只是死机？
	if _, ok := t.peers[id]; !ok {
		return
	}
	urls, err := types.NewURLs(us)
	if err != nil {
		if t.Logger != nil {
			t.Logger.Panic("failed NewURLs", zap.Strings("urls", us), zap.Error(err))
		}
	}
	t.peers[id].update(urls)

	t.pipelineProber.Remove(id.String())
	addPeerToProber(t.Logger, t.pipelineProber, id.String(), us, RoundTripperNameSnapshot, rttSec)
	t.streamProber.Remove(id.String())
	addPeerToProber(t.Logger, t.streamProber, id.String(), us, RoundTripperNameRaftMessage, rttSec)

	if t.Logger != nil {
		t.Logger.Info(
			"updated remote peer",
			zap.String("local-member-id", t.ID.String()),
			zap.String("updated-remote-peer-id", id.String()),
			zap.Strings("updated-remote-peer-urls", us),
		)
	}
}

func (t *Transport) ActiveSince(id types.ID) time.Time {
	t.mu.RLock()
	defer t.mu.RUnlock()
	if p, ok := t.peers[id]; ok {
		return p.activeSince()
	}
	return time.Time{}
}

// Transport. SendSnapshotO方法负责发送指定的snap.Message消息（其中封装了对应的MsgSnap消息实例及其他相关信息〉 。
// 该方法是通过调用peer.sendSnap（）方法完成的，
func (t *Transport) SendSnapshot(m snap.Message) {
	t.mu.Lock()
	defer t.mu.Unlock()
	p := t.peers[types.ID(m.To)]
	if p == nil {
		m.CloseWithError(errMemberNotFound)
		return
	}
	p.sendSnap(m)
}

// Pausable是用于暂停传输流量的测试接口。
type Pausable interface {
	Pause()
	Resume()
}

func (t *Transport) Pause() {
	t.mu.RLock()
	defer t.mu.RUnlock()
	for _, p := range t.peers {
		p.(Pausable).Pause()
	}
}

func (t *Transport) Resume() {
	t.mu.RLock()
	defer t.mu.RUnlock()
	for _, p := range t.peers {
		p.(Pausable).Resume()
	}
}

// ActivePeers返回通道l当初始
// 对等连接建立时关闭。使用此选项等待
// 第一个对等连接激活。
func (t *Transport) ActivePeers() (cnt int) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	for _, p := range t.peers {
		if !p.activeSince().IsZero() {
			cnt++
		}
	}
	return cnt
}
